﻿#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <vector>
#include <queue>
#include <memory>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <unordered_map>
#include <thread>
#include <future>
#include <iostream>

const int TASK_MAX_THRESHHODE = INT32_MAX;
const int THREAD_MAX_THRESHHODE = 512;
const int THREAD_MAX_IDEL_TIME = 60;  // 单位：秒

// 线程池的工作模式
enum class PoolMode
{
	MODE_FIXED,  // 固定数量的线程池
	MODE_CACHED, // 线程数量可动态变化的线程池
};

// 线程类型
class Thread
{
public:
	// 线程函数对象类型
	using ThreadFunc = std::function<void(int)>;

	// 线程构造
	Thread(ThreadFunc func)
		: func_(func)
		, threadId_(generateId_++)
	{
	}
	// 线程析构
	~Thread() = default;
	// 启动线程
	void start()
	{
		// 要执行一个线程函数
		std::thread t(func_, threadId_);
		t.detach();
	}
	// 获得线程ID
	int getId() const
	{
		return threadId_;
	}

private:
	ThreadFunc func_;
	static int generateId_;
	int threadId_;
};
int Thread::generateId_ = 0;

// 线程池类型
class ThreadPool
{
public:
	// 线程池构造
	ThreadPool()
		: initThreadSize_(0)
		, taskSize_(0)
		, idleThreadSize_(0)
		, curThreadSize_(0)
		, threadSizeThreshHold_(THREAD_MAX_THRESHHODE)
		, taskQueMaxThreshHold_(TASK_MAX_THRESHHODE)
		, poolMode_(PoolMode::MODE_FIXED)
		, isPoolRunning_(false)
	{
	}

	// 线程池析构
	~ThreadPool()
	{
		isPoolRunning_ = false;

		// 等待线程池中所有的线程返回
		std::unique_lock<std::mutex> lock(taskQueMtx_);
		notEmpty_.notify_all();
		exitCond_.wait(lock, [&]() ->bool { return threads_.size() == 0; });
	}


	// 开启线程池
	void start(int initThreadSize = std::thread::hardware_concurrency())
	{
		isPoolRunning_ = true;

		// 记录初始线程数量
		initThreadSize_ = initThreadSize;
		curThreadSize_ = initThreadSize;

		// 创建线程对象
		for (int i = 0; i < initThreadSize; ++i)
		{
			// 创建新线程
			auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
			// threads_.emplace_back(std::move(ptr));
			int threadId = ptr->getId();
			threads_.emplace(threadId, std::move(ptr));
		}

		// 启动所有线程
		for (int i = 0; i < initThreadSize; ++i)
		{
			threads_[i]->start(); // 去执行线程函数
			idleThreadSize_++;    // 记录初始空闲线程的数量
		}
	}

	// 设置线程池模式
	void setMode(PoolMode mode)
	{
		if (checkRunningState())
			return;

		poolMode_ = mode;
	}

	// 设置线程池任务队列最大数量
	void setTaskQueMaxThreshHold(int threshhold)
	{
		if (checkRunningState())
			return;

		taskQueMaxThreshHold_ = threshhold;
	}

	// 给线程池提交任务
	// 使用可变参模版编程，让submitTask可以接收任意任务函数和任意数量的参数
	// pool.submitTask(sum1, 10, 20, 30);
	// Result submitTask(std::shared_ptr<Task> sp);
	template<typename Func, typename... Args>
	auto submitTask(Func&& func, Args&&... args) -> std::future<decltype(func(args...))>  // 3个&符号，折叠出来是左值引用， 4个&符号，折叠出来是右值引用
	// func有可能是左值，有可能是右值
	{
		// 打包任务，放入任务队列中
		using RType = decltype(func(args...));
		auto task = std::make_shared<std::packaged_task<RType()>>(
			std::bind(std::forward<Func>(func), std::forward<Args>(args)...));
		std::future<RType> result = task->get_future();

		// 获取锁
		std::unique_lock<std::mutex> lock(taskQueMtx_);

		// 如果阻塞了1s还没等到，直接返回，提示用户任务队列已满，稍后操作
		if (!notFull_.wait_for(lock, std::chrono::seconds(1),
			[&]() ->bool { return taskQue_.size() < (size_t)taskQueMaxThreshHold_; }))
		{
			std::cerr << "task queue is full, submit task fail." << std::endl;
			auto task = std::make_shared<std::packaged_task<RType()>>(
				[]() -> RType { return RType(); });
			(*task)();
			return task->get_future();
		}

		// 如果有空余，把任务放入任务队列中
		// using Task = std::function<void()>;
		taskQue_.emplace([task]() { (*task)(); });
		taskSize_++;

		// 因为放了新任务，任务队列一定不空了，在notEmpty_上通知，分配线程执行任务
		notEmpty_.notify_all();

		// cached模式，任务处理比较紧急，场景：小而快
		if (poolMode_ == PoolMode::MODE_CACHED
			&& taskSize_ > idleThreadSize_
			&& curThreadSize_ < threadSizeThreshHold_)
		{
			// 创建新线程
			auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
			// threads_.emplace_back(std::move(ptr));
			int threadId = ptr->getId();
			threads_.emplace(threadId, std::move(ptr));
			// 启动线程
			threads_[threadId]->start();
			// 修改相关变量
			curThreadSize_++;
			idleThreadSize_++;

			std::cout << ">>> create new thread successful" << std::endl;
		}

		// 返回任务的Result对象
		return result;

	}
	// 线程函数
	void threadFunc(int threadId)
	{
		auto lastTime = std::chrono::steady_clock().now();
		for (;;)
		{
			// std::shared_ptr<Task> task = nullptr;
			Task task;
			{
				// 获取锁
				std::unique_lock<std::mutex> lock(taskQueMtx_);
				std::cout << std::this_thread::get_id() << "尝试获取任务" << std::endl;

				// 超过initThreadSize_的线程要进行回收

				while (taskQue_.size() == 0)
				{
					//// 线程池要回收线程资源
					if (!isPoolRunning_)
					{
						threads_.erase(threadId);
						curThreadSize_--;
						idleThreadSize_--;
						std::cout << std::this_thread::get_id() << " exit!" << std::endl;
						exitCond_.notify_all();
						return;
					}
					if (poolMode_ == PoolMode::MODE_CACHED)
					{
						// 条件变量超时返回
						if (std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
						{
							auto now = std::chrono::steady_clock().now();
							auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
							if (dur.count() >= THREAD_MAX_IDEL_TIME && curThreadSize_ > initThreadSize_)
							{
								threads_.erase(threadId);
								curThreadSize_--;
								idleThreadSize_--;
								// 问题：ThreadFunc 是被哪个 Thread执行   
								std::cout << std::this_thread::get_id() << " exit!" << std::endl;
								return;
							}
						}
					}
					else
					{
						// 等待 notEmpty 条件
						notEmpty_.wait(lock);
					}
				}

				idleThreadSize_--;

				// 从任务队列中取一个任务出来
				task = taskQue_.front();
				taskQue_.pop();
				taskSize_--;

				std::cout << std::this_thread::get_id() << "获取任务成功" << std::endl;

				// 如果依然有剩余任务，继续通知其他线程执行任务
				if (taskQue_.size() > 0)
				{
					notEmpty_.notify_all();
				}

				// 取出一个任务，进行通知，通知可以继续生产任务
				notFull_.notify_all();
			} // 保证释放锁之后再执行任务

			// 当前线程执行这个任务
			if (task != nullptr)
			{
				// task->run(); // 执行任务，把任务的返回值通过setVal接口返回给Result
				// task->exec();
				task();
			}
			lastTime = std::chrono::steady_clock().now();
			idleThreadSize_++;
		}
	}

	// 设置线程数量的上限
	void setThreadMaxThreshHold(int threshhold)
	{
		if (checkRunningState())
			return;
		if (this->poolMode_ != PoolMode::MODE_CACHED)
			return;
		threadSizeThreshHold_ = threshhold;
	}

	ThreadPool(const ThreadPool&) = delete;
	ThreadPool& operator=(const ThreadPool&) = delete;

public:
	// 检查线程池的运行状态
	bool checkRunningState() const
	{
		return isPoolRunning_;
	}

private:

	// std::vector<std::unique_ptr<Thread>> threads_; // 线程列表
	std::unordered_map<int, std::unique_ptr<Thread>> threads_;
	int initThreadSize_;                           // 初始的线程数量
	int threadSizeThreshHold_;                     // 线程数量的上限阈值
	std::atomic_int curThreadSize_;                // 记录当前线程池中的线程个数
	std::atomic_int idleThreadSize_;               // 记录空闲线程的数据量

	// Task任务 -> 函数对象
	using Task = std::function<void()>;
	std::queue<Task> taskQue_;                  // 任务队列
	std::atomic_int taskSize_;                  // 任务数量
	int	taskQueMaxThreshHold_;                  // 任务队列上限阈值

	std::mutex taskQueMtx_;                     // 保证任务队列的线程安全
	std::condition_variable notFull_;           // 任务队列不满
	std::condition_variable notEmpty_;          // 任务队列不空
	std::condition_variable exitCond_;

	PoolMode poolMode_;              // 线程池的工作模式
	std::atomic_bool isPoolRunning_; // 表示线程池的启动状态
};

#endif